package storm.cookbook;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;
import storm.constants.Constants;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

//import backtype.storm.task.OutputCollector;
//import backtype.storm.topology.IRichSpout;
//import backtype.storm.topology.base.BaseRichBolt;

public class HelloWorldSplitBolt extends BaseBasicBolt {

    /**
     * 创建一个计数用的容器！
     */
    private Map<String, Integer> countMap;


    @SuppressWarnings("rawtypes")
    @Override
    public void prepare(Map stormConf, TopologyContext context) {
        System.out.println("创建HelloWorldSplitBolt 的实例 进行处理 bolt");
        // 在第一次初始化的时候进行 初始化
        countMap = new HashMap<>();

        countMap.put("heihei", 0);
        countMap.put("haha", 0);
        countMap.put("lalala", 0);
        countMap.put("other", 0);

    }

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        // 通过这个方法进行获取对应的参数
        List list = (List) input.getValueByField(Constants.fieldType.word.toString());
        for (Object s : list) {
            String word = (String) s;
            // reduct 的流程
            switch (word) {
                case "heihei":
                    countMap.put("heihei", countMap.get("heihei") + 1);
                    break;

                case "haha":
                    countMap.put("haha", countMap.get("haha") + 1);
                    break;

                case "lalala":
                    countMap.put("lalala", countMap.get("lalala") + 1);
                    break;

                default:
                    countMap.put("other", countMap.get("other") + 1);
                    break;
            }
        }
        System.out.println("countMap 统计的数字:  heihei: " + countMap.get("heihei") + "  haha: " + countMap.get("haha") + " lalala: " + countMap.get("lalala"));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }


}
